package mysql;

import com.github.shyiko.mysql.binlog.BinaryLogFileReader;
import com.github.shyiko.mysql.binlog.event.*;
import com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer;
import lombok.extern.slf4j.Slf4j;

import java.io.File;
import java.io.Serializable;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import static com.github.shyiko.mysql.binlog.event.EventType.*;

@Slf4j
public class MysqlBinlog {

    // 接下来的事件操作的库和表 .连接
    private static String dbTable;
    // 数据库表-字段信息
    private static Map<String, Map<String, Colum>> dbTableCols = new ConcurrentHashMap<>();

    // binlog解析Demo：通过mysql-binlog-connector-java解析binlog二进制文件
    // https://www.codeleading.com/article/16432296956/
    // Java监听mysql的binlog详解(mysql-binlog-connector)
    // https://blog.csdn.net/m0_37583655/article/details/119148470
    public static void main(String[] args) throws Exception {
        // 获取表-字段信息
        init("demo1", "t_sys_user_role");

        String filePath = "D:\\softs\\mysql-5.7.33-winx64\\data\\mysql-bin.000002";
        File binlogFile = new File(filePath);
        EventDeserializer eventDeserializer = new EventDeserializer();
        eventDeserializer.setCompatibilityMode(
                EventDeserializer.CompatibilityMode.DATE_AND_TIME_AS_LONG,
                EventDeserializer.CompatibilityMode.CHAR_AND_BINARY_AS_BYTE_ARRAY
        );
        BinaryLogFileReader reader = new BinaryLogFileReader(binlogFile, eventDeserializer);
        try {
            for (Event event; (event = reader.readEvent()) != null; ) {
                System.out.println(event.toString());
                // 获取事件类型 Mysql Binlog event类型
                EventType eventType = event.getHeader().getEventType();

                System.out.println("事件类型: " + eventType);

                // MySQL Binlog日志 http://www.qb5200.com/article/542539.html
                // Binlog是Binary log的缩写，即二进制日志。Binlog主要有三个作用：持久化时将随机IO转化为顺序IO，主从复制以及数据恢复。本文重点主从复制相关的问题。
                // Binlog日志由一个索引文件与很多日志文件组成，每个日志文件由魔数以及事件组成，每个日志文件都会以一个Rotate类型的事件结束。

                // 230_mysql_binlog结构组成（mysql binlog详解） https://www.huoban.com/news/post/19452.html
                // 在解析之前，首先要多binlog的事件类型有一个大致了解，不过我们只关注row模式
                // 获取接下来要操作的库名、表名
                // MySQL binlog 格式解析(TABLE_MAP_EVENT) https://baijiahao.baidu.com/s?id=1649870303802210611
                // row格式的binlog文件中，每个ROW_EVENT之前都有一个TABLE_MAP_EVENT，用于描述表的内部ID和结构定义
                // show binlog events in "mysql-bin.000002";

                // TABLE_MAP_EVENT只有在binlog文件是以ROW格式记录的时候，才会使用。binlog中记录的每个更改的记录之前都会有一个对应要操作的表的TABLE_MAP_EVENT。
                // TABLE_MAP_EVENT中记录了表的定义（包括database name,table name，字段定义），
                // 并且会将这个表的定义对应于一个数字，称为table_id。设计TABLE_MAP_EVENT类型event的目的是为了当主库和从库之间有不同的表定义的时候，
                // 复制仍能进行。如果一个事务中操作了多个表，多行记录，在binlog中会将对多行记录的操作event进行分组，
                // 每组行记录操作event前面会出现对应表的TABLE_MAP_EVENT。

                // ROWS_EVENT
                // 对于ROW格式的binlog，所有的DML语句都是记录在ROWS_EVENT中。
                // ROWS_EVENT分为三种：WRITE_ROWS_EVENT，UPDATE_ROWS_EVENT，DELETE_ROWS_EVENT
                // 对于QUERY_EVENT事件，是以文本形式记录DML操作的。而对于ROWS_EVENT事件，并不是文本形式，所以在通过mysqlbinlog查看基于ROW格式的binlog时，需要指定-vv --base64-output=decode-rows
                // WRITE_ROWS_EVENT
                // UPDATE_ROWS_EVENT
                // DELETE_ROWS_EVENT
                // 在以ROW格式记录的binlog文件中，WRITE_ROWS_EVENT记录了插入的行记录
                // 在以ROW格式记录的binlog文件中，UPDATE_ROWS_EVENT记录了更新的行记录
                // 在以ROW格式记录的binlog文件中，DELETE_ROWS_EVENT记录了删除的行记录

                // 【MySQL（二十一）】binlog 事件 https://blog.csdn.net/u010900754/article/details/108458028
                // binlog事件类型
                // 总共有3个版本：v1，v2和v4。mysql5以上使用过的v4。这里只记录这几个关键的事件类型：
                // QUERY_EVENT：在statement模式下，增删改的语句都会生成该事件；在row模式下，DDL的改动会生成该事件;
                // ROTATE_EVENT：新的binlog文件生成时，会记录该事件，内容就是下一个binlog文件的文件名；
                // FORMAT_DESCRIPTION_EVENT：每一个binlog文件的起始事件，描述文件属性；
                // TABLE_MAP_EVENT：在row模式下会有，每一个更新事件都会先有一个TABLE_MAP_EVENT事件，用于记录表的一些信息。
                // WRITE_ROWS_EVENT：在row模式下会有，insert；
                // UPDATE_ROWS_EVENT：在row模式下会有，update；
                // DELETE_ROWS_EVENT：在row模式下会有，delete；

                // binlog格式可以在线修改：set global binlog_format='statement/row/mixed';
                // 查看binlog格式：show variables like '%binlog%';

                // Binlog 事件的结构 https://juejin.cn/post/6844903794073960455
                // 一个事件对象分为事件头和事件体，事件的结构如下：
                // event header   type_code
                // event data

                if (eventType == EventType.TABLE_MAP) {
                    // TABLE_MAP_EVENT：在row模式下会有，每一个更新事件都会先有一个TABLE_MAP_EVENT事件，用于记录表的一些信息。
                    TableMapEventData tableData = event.getData();
                    String db = tableData.getDatabase();
                    String table = tableData.getTable();
                    dbTable = getdbTable(db, table);
                }

                if (eventType == QUERY) {
                    // QUERY_EVENT：在statement模式下，增删改的语句都会生成该事件；在row模式下，DDL的改动会生成该事件
                    // DDL（data definition language）是数据定义语言：DDL比DML要多，主要的命令有CREATE、ALTER、DROP等，DDL主要是用在定义或改变表（TABLE）的结构，数据类型，表之间的链接和约束等初始化工作上，他们大多在建立表时使用。
                    // DML（data manipulation language）是数据操纵语言：它们是SELECT、UPDATE、INSERT、DELETE，就象它的名字一样，这4条命令是用来对数据库里的数据进行操作的语言。
                    QueryEventData queryEventData = event.getData();
                    System.out.println(queryEventData);
                }


                // 根据操作类型获取对应集合
                // MysqlBinLogListener 数据库监听器
                // 只处理添加删除更新三种操作
                if (isWrite(eventType) || isUpdate(eventType) || isDelete(eventType)) {
                    if (isWrite(eventType)) {
                        WriteRowsEventData data = event.getData();
                        for (Serializable[] row : data.getRows()) {
                            BinLogItem item = BinLogItem.itemFromInsertOrDeleted(row, dbTableCols.get(dbTable), eventType);
                            item.setDbTable(dbTable);
                            System.out.println(item);
                        }
                    }
                    if (isUpdate(eventType)) {
                        UpdateRowsEventData data = event.getData();
                        for (Map.Entry<Serializable[], Serializable[]> row : data.getRows()) {
                            if (dbTableCols.containsKey(dbTable)) {
                                BinLogItem item = BinLogItem.itemFromUpdate(row, dbTableCols.get(dbTable), eventType);
                                item.setDbTable(dbTable);
                                System.out.println(item);
                            }
                        }

                    }
                    if (isDelete(eventType)) {
                        DeleteRowsEventData data = event.getData();
                        for (Serializable[] row : data.getRows()) {
                            if (dbTableCols.containsKey(dbTable)) {
                                BinLogItem item = BinLogItem.itemFromInsertOrDeleted(row, dbTableCols.get(dbTable), eventType);
                                item.setDbTable(dbTable);
                                System.out.println(item);
                            }
                        }
                    }
                }
            }
        } finally {
            reader.close();
        }
    }

    /**
     * 拼接dbTable
     */
    public static String getdbTable(String db, String table) {
        return db + "@" + table;
    }

    /**
     * 查询数据库表-字段信息
     *
     * @param db       数据库
     * @param table    操作表
     */
    public static void init(String db, String table) throws Exception {
        String dbTable = getdbTable(db, table);
        // 获取字段集合
        Map<String, Colum> cols = getColMap(db, table);
        // 保存字段信息
        dbTableCols.put(dbTable, cols);
    }

    /**
     * 获取columns集合
     */
    public static Map<String, Colum> getColMap(String db, String table) throws ClassNotFoundException {
        try {
            Class.forName("com.mysql.jdbc.Driver");
            // 保存当前注册的表的colum信息
            // Connection connection = DriverManager.getConnection("jdbc:mysql://" + conf.getHost() + ":" + conf.getPort(), conf.getUsername(), conf.getPasswd());
            Connection connection = DriverManager.getConnection("jdbc:mysql://127.0.0.1:3306", "root", "123456");
            // 执行sql
            String preSql = "SELECT TABLE_SCHEMA, TABLE_NAME, COLUMN_NAME, DATA_TYPE, ORDINAL_POSITION FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = ? and TABLE_NAME = ?";
            PreparedStatement ps = connection.prepareStatement(preSql);
            ps.setString(1, db);
            ps.setString(2, table);
            ResultSet rs = ps.executeQuery();
            Map<String, Colum> map = new HashMap<>(rs.getRow());
            while (rs.next()) {
                String schema = rs.getString("TABLE_SCHEMA");
                String tableName = rs.getString("TABLE_NAME");
                String column = rs.getString("COLUMN_NAME");
                int idx = rs.getInt("ORDINAL_POSITION");
                String dataType = rs.getString("DATA_TYPE");
                if (column != null && idx >= 1) {
                    map.put(column, new Colum(schema, tableName, idx - 1, column, dataType)); // sql的位置从1开始
                }
            }
            ps.close();
            rs.close();
            return map;
        } catch (Exception e) {
            log.error("load db conf error, db_table={}:{} ", db, table, e);
        }
        return null;
    }

}
